import httpx
import json
from typing import Any
from mcp.server.fastmcp import FastMCP
import requests
import re

# Initialize FastMCP server
mcp = FastMCP("mapServer")

# API配置
GAODE_ROUTE_API = ""
USER_AGENT = "map-app/1.0"
API_KEY = ""

# 坐标格式验证
COORD_PATTERN = re.compile(r"^\d{1,3}\.\d+,\d{1,3}\.\d+$")
def validate_coordinates(coord: str) -> bool:
    """验证坐标格式是否符合要求"""
    if not COORD_PATTERN.match(coord):
        return False
    lon, lat = map(float, coord.split(','))
    return -180 <= lon <= 180 and -90 <= lat <= 90

async def get_navigation(start_point: str, destination: str):
    """
    导航路线查询功能
    :param start_point: 起点坐标(格式:经度,纬度)
    :param destination: 终点坐标(格式:经度,纬度)
    :return: 导航结果
    """
    # 验证坐标格式
    if not validate_coordinates(start_point):
        return {"status": "0", "info": f"无效起点坐标格式: {start_point}"}
    if not validate_coordinates(destination):
        return {"status": "0", "info": f"无效终点坐标格式: {destination}"}

    params = {
        "key": API_KEY,
        "origin": start_point,
        "destination": destination,
        "output": "json"
    }
    headers = {"User-Agent": USER_AGENT}

    try:
        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.get(
                GAODE_ROUTE_API,
                params=params,
                headers=headers
            )
            response.raise_for_status()
            return response.json()
    except httpx.HTTPStatusError as e:
        return {"status": "error", "info": f"HTTP错误: {e.response.status_code}"}
    except httpx.TimeoutException:
        return {"status": "error", "info": "请求超时"}
    except Exception as e:
        return {"status": "error", "info": f"请求失败: {str(e)}"}


def format_navigation(data: dict[str, Any]) -> str:
    """
    格式化导航结果
    :param data: API响应数据
    :return: 格式化后的字符串
    """
    # 处理客户端错误
    if data.get("status") == "error":
        return f"⚠️ 请求错误: {data.get('info', '未知错误')}"

    if data.get("status") != "1":
        error_info = data.get("info", "未知错误")
        return f"⚠️ API请求失败: {error_info} (状态码: {data.get('status')})"

    route = data.get("route", {})
    paths = route.get("paths", [])
    if not paths:
        return "未找到导航路线"

    path = paths[0]  # 取第一条路线
    result = ["🚗 导航路线方案:"]

    # 添加关键路线信息
    distance = int(path.get("distance", 0))
    duration = int(path.get("duration", 0))
    tolls = float(path.get("tolls", 0))

    result.append(f"📏 总距离: {distance / 1000:.1f}公里 ({distance}米)")
    result.append(f"⏱️ 预计时间: {duration // 60}分钟 ({duration}秒)")
    result.append(f"💰 收费路段: {tolls:.2f}元")
    result.append(f"🚦 红绿灯数: {path.get('traffic_lights', '未知')}")

    # 添加详细步骤
    steps = path.get("steps", [])
    if steps:
        result.append("\n🗺️ 路线详情:")
        for i, step in enumerate(steps[:5], 1):  # 只显示前5个步骤
            instruction = step.get("instruction", "").replace("<b>", "").replace("</b>", "")
            result.append(f"{i}. {instruction}")
            step_dist = int(step.get("distance", 0))
            result.append(f"   ➡️ 距离: {step_dist}米 | 方向: {step.get('orientation', '未知')}")

        if len(steps) > 5:
            result.append(f"\n... 省略{len(steps) - 5}个步骤 ...")

    return "\n".join(result)

@mcp.tool()
async def query_navigation(origin: str, destination: str) -> str:
    """
    导航查询工具
    :param origin: 起点坐标(经度,纬度)
    :param destination: 终点坐标(经度,纬度)
    :return: 格式化后的导航信息
    """
    data = await get_navigation(origin, destination)
    return format_navigation(data)

if __name__ == '__main__':
    # mcp.run(transport='stdio')
    import asyncio

    async def test_navigation():
        test_cases = [
            ("116.481028,39.989643", "116.465302,40.004717"),  # 有效坐标
            ("invalid_coord", "116.465302,40.004717"),  # 无效起点
            ("116.481028,39.989643", "invalid_coord"),  # 无效终点
            ("200.000,100.000", "116.465302,40.004717")  # 越界坐标
        ]

        for origin, dest in test_cases:
            print(f"\n测试导航: {origin} → {dest}")
            result = await query_navigation(origin, dest)
            print(result)
            print("-" * 50)

    asyncio.run(test_navigation())